﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using UtilZ.Dotnet.Ex.Base;
using UtilZ.Dotnet.Ex.Exceptions;
using UtilZ.Dotnet.Ex.Log;
using UtilZ.Dotnet.Ex.ZLog;

namespace UtilZ.Dotnet.Ex.DataStruct
{
    /// <summary>
    /// 异步队列
    /// </summary>
    /// <typeparam name="T">数据类型</typeparam>
    public class AsynQueueX<T> : IDisposableZ
    {
        /// <summary>
        /// 队列消费者线程
        /// </summary>
        private readonly ThreadEx _thread = null;

        /// <summary>
        /// 获取队列线程状态[true:线程正在运行;false:线程未运行]
        /// </summary>
        public bool Status
        {
            get { return this._thread.IsRuning; }
        }

        /// <summary>
        /// 默认值
        /// </summary>
        private readonly T _defaultValue;

        /// <summary>
        /// 数组集合
        /// </summary>
        private T[] _array;

        /// <summary>
        /// 队列容量
        /// </summary>
        private readonly int _capcity;

        /// <summary>
        /// 集合项数
        /// </summary>
        private int _count = 0;

        /// <summary>
        /// 集合项数下限值,-1
        /// </summary>
        private readonly int _countLowerLimit = -1;

        /// <summary>
        /// 集合项数上限值
        /// </summary>
        private int _countUpperLimit;

        /// <summary>
        /// 集合写索引
        /// </summary>
        private int _writeIndex = 0;

        /// <summary>
        /// 集合读索引
        /// </summary>
        private int _readIndex = 0;

        /// <summary>
        /// true:有容量限制;false:无容量限制
        /// </summary>
        private bool _hasCapcity;



        /// <summary>
        /// 选项
        /// </summary>
        private readonly AsynQueueOptions _options;

        /// <summary>
        /// 空队列等待线程消息通知
        /// </summary>
        private readonly AutoResetEvent _emptyQueueWaitEventHandle = new AutoResetEvent(false);

        private void EmptyQueueWaitEventHandleSet()
        {
            try
            {
                this._emptyQueueWaitEventHandle.Set();
            }
            catch (ObjectDisposedException)
            { }
        }


        /// <summary>
        /// 对象是否已释放[true:已释放;false:未释放]
        /// </summary>
        private bool _disposed = false;
        /// <summary>
        /// 获取当前对象是否已释放[true:已释放;false:未释放]
        /// </summary>
        bool IDisposableZ.Disposed
        {
            get { return _disposed; }
        }

        /// <summary>
        /// 数据输出类型
        /// </summary>
        private readonly OutputType _outputType;

        /// <summary>
        /// 数据处理委托
        /// </summary>
        private readonly Action<T> _processSingleItemAction;

        /// <summary>
        /// 数据处理委托[第一个参数：元素数组；第二个参数有效元素个数]
        /// </summary>
        private readonly Action<T[], int> _processCollectionAction;

        /// <summary>
        /// 容量计数信号量
        /// </summary>
        private readonly SemaphoreSlim _semaphoreSlim;



        /// <summary>
        /// 构造函数
        /// </summary>
        /// <param name="outputType">输出类型</param>
        /// <param name="options">选项</param>
        private AsynQueueX(OutputType outputType, AsynQueueOptions options)
        {
            if (options.Capcity < 1)
            {
                throw new ArgumentOutOfRangeException(nameof(options.Capcity));
            }

            this._outputType = outputType;
            this._capcity = options.Capcity;
            int size = this._capcity;
            if (size == int.MaxValue)
            {
                size = 100;//初始容量
                this._hasCapcity = false;
            }
            else
            {
                this._hasCapcity = true;
            }

            this._array = new T[size];
            this._countUpperLimit = this._capcity + 1;
            this._defaultValue = default(T);

            this._options = options;
            this._semaphoreSlim = new SemaphoreSlim(this._capcity);
            this._thread = new ThreadEx(this.RunThreadQueueProcessMethod, options.ThreadName, options.IsBackground);
            if (options.AutoStart)
            {
                this.Start();
            }
        }

        /// <summary>
        /// 构造函数-单项输出
        /// </summary>
        /// <param name="processAction">数据处理委托</param>
        /// <param name="options">选项</param>
        public AsynQueueX(Action<T> processAction, AsynQueueOptions options) :
            this(OutputType.Single, options)
        {
            this._processSingleItemAction = processAction;
        }



        /// <summary>
        /// 构造函数-批量输出
        /// </summary>
        /// <param name="processAction">数据处理委托[第一个参数：元素数组；第二个参数有效元素个数]</param>
        /// <param name="options">选项</param>
        public AsynQueueX(Action<T[], int> processAction, AsynQueueOptions options) :
            this(OutputType.Batch, options)
        {
            if (options.BatchCount < 1)
            {
                throw new ArgumentException(string.Format("批量处理最大项数不能小于1,值:{0}无效", options.BatchCount));
            }

            this._processCollectionAction = processAction;
        }





        /// <summary>
        /// 启动子类无参数工作线程
        /// </summary>
        public void Start()
        {
            if (this._disposed)
            {
                throw new ObjectDisposedException(this.ToString());
            }

            this._thread.Start();
        }

        /// <summary>
        /// 停止工作线程
        /// </summary>       
        /// <param name="sync">是否同步停止[true:同步停止;false:异常停止];注:注意线程死锁,典型场景:刷新UI,在UI上执行同步停止</param>
        /// <param name="synMillisecondsTimeout">同步超时时间,-1表示无限期等待,单位/毫秒[isSycn为true时有效]</param>
        public void Stop(bool sync = false, int synMillisecondsTimeout = -1)
        {
            if (sync)
            {
                this._thread.SyncStop(synMillisecondsTimeout);
            }
            else
            {
                this._thread.Stop();
            }
            this.EmptyQueueWaitEventHandleSet();
        }

        /// <summary>
        /// 线程队列处理方法
        /// </summary>
        private void RunThreadQueueProcessMethod(ThreadPara threadPara)
        {
            CancellationToken token = threadPara.Token;
            switch (this._outputType)
            {
                case OutputType.Batch:
                    this.BatchOutputProcess(token);
                    break;
                case OutputType.Single:
                    this.SingleOutputProcess(token);
                    break;
                default:
                    throw new NotImplementedException(this._outputType.ToString());
            }
        }

        /// <summary>
        /// 单项输出处理
        /// </summary>
        private void SingleOutputProcess(CancellationToken token)
        {
            try
            {
                T item;
                while (!token.IsCancellationRequested)
                {
                    try
                    {
                        if (this.TryDequeue(out item))
                        {
                            //数据处理                            
                            this._processSingleItemAction?.Invoke(item);
                        }
                        else
                        {
                            try
                            {
                                this._emptyQueueWaitEventHandle.WaitOne(this._options.EmptyQueueWaitTimeout);
                            }
                            catch (ObjectDisposedException)
                            {
                                continue;
                            }
                        }
                    }
                    catch (Exception exi)
                    {
                        ZLog.ZLoger.Error(exi);
                    }
                }
            }
            catch (Exception ex)
            {
                ZLog.ZLoger.Error(ex);
            }
        }

        /// <summary>
        /// 批量输出处理
        /// </summary>
        private void BatchOutputProcess(CancellationToken token)
        {
            try
            {
                int batchCount = this._options.BatchCount;
                T[] arr = new T[batchCount];
                int index, count;

                while (!token.IsCancellationRequested)
                {
                    try
                    {
                        count = 0;
                        index = 0;

                        while (count < batchCount && !token.IsCancellationRequested)
                        {
                            if (this.TryDequeue(out arr[index]))
                            {
                                index++;
                                count++;
                            }
                            else
                            {
                                try
                                {
                                    if (this._emptyQueueWaitEventHandle.WaitOne(this._options.BatchMillisecondsTimeout))
                                    {
                                        continue;
                                    }
                                    else
                                    {
                                        break;
                                    }
                                }
                                catch (ObjectDisposedException)
                                {
                                    continue;
                                }
                            }
                        }

                        if (count > 0 && !token.IsCancellationRequested)
                        {
                            //数据处理
                            this._processCollectionAction?.Invoke(arr, count);//数据处理
                        }
                        else
                        {
                            try
                            {
                                this._emptyQueueWaitEventHandle.WaitOne(this._options.EmptyQueueWaitTimeout);
                            }
                            catch (ObjectDisposedException)
                            {
                                continue;
                            }
                        }
                    }
                    catch (Exception exi)
                    {
                        ZLog.ZLoger.Error(exi);
                    }
                }
            }
            catch (Exception ex)
            {
                ZLog.ZLoger.Error(ex);
            }
        }


        /// <summary>
        /// 尝试出队列一项,出成功返回true,失败返回false
        /// </summary>
        /// <param name="item">输出项</param>
        /// <returns>出成功返回true,失败返回false</returns>
        public bool TryDequeue(out T item)
        {
            int count = Interlocked.Decrement(ref this._count);
            if (count < 0)
            {
                Interlocked.CompareExchange(ref this._count, 0, this._countLowerLimit);
                item = this._defaultValue;
                return false;
            }

            item = this._array[this._readIndex];
            Interlocked.Increment(ref this._readIndex);
            int readIndex = Interlocked.CompareExchange(ref this._readIndex, 0, this._capcity);
            this._semaphoreSlim.Release();
            return true;
        }


        /// <summary>
        /// 出队列一项
        /// </summary>
        /// <returns></returns>
        /// <exception cref="Exception"></exception>
        public T Dequeue()
        {
            T result;
            if (this.TryDequeue(out result))
            {
                return result;
            }
            else
            {
                throw new Exception("没有可出队的项");
            }
        }

        /// <summary>
        /// 尝试获取队列头项,但不移除,获取成功返回true,失败返回false
        /// </summary>
        /// <param name="item">输出项</param>
        /// <returns>获取成功返回true,失败返回false</returns>
        public bool TryPeek(out T item)
        {
            if (this._count <= 0)
            {
                item = this._defaultValue;
                return false;
            }

            item = this._array[this._readIndex];
            return true;
        }



        /// <summary>
        /// 将对象添加到队列的结尾处
        /// </summary>
        /// <param name="item">待添加的对象</param>
        public void Enqueue(T item)
        {
            if (this._disposed)
            {
                throw new ObjectDisposedException(this.ToString());
            }

            if (this._hasCapcity)
            {
                int count1, count2;
                while (true)
                {
                    count1 = Interlocked.Increment(ref this._count);
                    if (count1 > this._capcity)
                    {
                        count2 = Interlocked.CompareExchange(ref this._count, this._capcity, this._countUpperLimit);
                        if (count2 == count1)
                        {
                            //两个值相等,说明没有其它线程在此时改变值
                            switch (this._options.FullMode)
                            {
                                case ContainerFullMode.Exception:
                                    if (this._semaphoreSlim.Wait(this._options.EnqueueMillisecondsTimeout))
                                    {
                                        count1 = 0;
                                        count2 = 1;
                                    }
                                    else
                                    {
                                        throw new TimeoutException("添加项到队列超时");
                                    }
                                    break;
                                case ContainerFullMode.DropNew:
                                    count1 = Interlocked.Decrement(ref this._writeIndex);
                                    count2 = Interlocked.CompareExchange(ref this._writeIndex, this._array.Length - 1, this._countLowerLimit);
                                    break;
                                case ContainerFullMode.DropOld:
                                    count1 = Interlocked.Increment(ref this._readIndex);
                                    count2 = Interlocked.CompareExchange(ref this._readIndex, 0, this._array.Length);
                                    break;
                                case ContainerFullMode.Ignore:
                                    //直接返回即可
                                    ZLoger.Warn($"{nameof(AsynQueueX<T>)}忽略数据项");
                                    return;
                                default:
                                    throw new NotImplementedException(this._options.FullMode.ToString());
                            }

                            if (count1 != count2)
                            {
                                continue;
                            }
                            else
                            {
                                break;
                            }
                        }
                        else
                        {
                            continue;
                        }
                    }
                    else
                    {
                        break;
                    }
                }
            }

            this.AppendItem(ref item);
            this.EmptyQueueWaitEventHandleSet();
        }


        private void AppendItem(ref T item)
        {
            int writeIndex = Interlocked.Increment(ref this._writeIndex);
            if (writeIndex >= this._array.Length)
            {
                if (this._count >= this._array.Length)
                {
                    if (this._count >= this._capcity)
                    {
                        throw new ContainerFullException();
                    }
                    else
                    {
                        long newLength;
                        newLength = (long)(this._array.Length * this._options.CapacityExpansionFactor);
                        if (newLength > int.MaxValue)
                        {
                            newLength = int.MaxValue;
                        }

                        var array = new T[(int)newLength];
                        Array.Copy(this._array, this._readIndex, array, 0, this._count);
                        this._array = array;
                        this._countUpperLimit = array.Length + 1;
                        this._readIndex = 0;
                        this._writeIndex = this._count;
                    }
                }
                else
                {
                    this._writeIndex = 0;
                }
            }
           
            writeIndex = Interlocked.CompareExchange(ref this._writeIndex, 0, this._capcity);
            this._array[writeIndex] = item;
        }




        /// <summary>
        /// 移除位于开始处的指定个数对象
        /// </summary>
        /// <param name="removeCount">要移除的项数</param>
        /// <returns>移除项集合</returns>
        public List<T> Remove(int removeCount)
        {
            if (this._disposed)
            {
                throw new ObjectDisposedException(this.ToString());
            }

            var removeItems = new List<T>();
            T item;
            int count;

            while (removeItems.Count < removeCount)
            {
                count = Interlocked.Decrement(ref this._count);
                if (count < 0)
                {
                    Interlocked.CompareExchange(ref this._count, 0, this._countLowerLimit);
                    break;
                }
                else
                {
                    item = this._array[this._readIndex];
                    this._semaphoreSlim.Release();
                    removeItems.Add(item);
                }
            }

            return removeItems;
        }

        /*
        /// <summary>
        /// 移除满足条件的元素
        /// </summary>
        /// <param name="predicate">用于定义要移除的元素应满足的条件</param>
        /// <returns>移除项集合</returns>
        public List<T> Remove(Func<T, bool> predicate)
        {
            if (predicate == null)
            {
                throw new ArgumentNullException(nameof(predicate));
            }

            var removeItems = new List<T>();
            if (this._array.Any(predicate))
            {
                T[] array = this._array.ToArray();
                this.PrimitiveClear();
                T item;
                for (int i = 0; i < array.Length; i++)
                {
                    item = array[i];
                    if (predicate(item))
                    {
                        removeItems.Add(item);
                    }
                    else
                    {
                        this.AppendItem(ref item);
                    }
                }
            }

            return removeItems;
        }*/

        /// <summary>
        /// 获取队列中包含的元素数
        /// </summary>
        public int Count
        {
            get
            {
                return this._count;
            }
        }

        /// <summary>
        /// 将队列中存储的元素复制到新数组中
        /// </summary>
        /// <returns>新数组</returns>
        public T[] ToArray()
        {
            var arr = new T[this._count];
            Array.Copy(this._array, this._readIndex, arr, 0, this._count);
            return arr;
        }

        /// <summary>
        /// 清空队列,必须在停止时执行,否则后果未知
        /// </summary>
        public void Clear()
        {
            this.PrimitiveClear();
        }


        private void PrimitiveClear()
        {
            this._count = 0;
            this._writeIndex = 0;
            this._readIndex = 0;
            for (int i = 0; i < this._array.Length; i++)
            {
                this._array[i] = this._defaultValue;
            }

            while (this._semaphoreSlim.CurrentCount < this._capcity)
            {
                try
                {
                    this._semaphoreSlim.Release();
                }
                catch (SemaphoreFullException)
                {
                    break;
                }
            }
        }

        /// <summary>
        /// 释放资源
        /// </summary>
        public void Dispose()
        {
            lock (this)
            {
                if (this._disposed)
                {
                    return;
                }
                this._disposed = true;

                this.Stop(false, 5000);

                this._thread.Dispose();
                this._emptyQueueWaitEventHandle.Dispose();
                this._semaphoreSlim.Dispose();
            }
        }


        private enum OutputType
        {
            Single,

            Batch
        }

    }

}
